package customerproportion.spark;

import customerproportion.User.UserInfo;
import customerproportion.User.UserLog;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.*;

public class GetCustomerProportion {
    public static void GetResult(Map<String, Long> ItemPairResult) {
        List<Map.Entry<String, Long>> entryList = new ArrayList<>(ItemPairResult.entrySet());
        entryList.sort(Map.Entry.comparingByKey());
        System.out.println(entryList.size());
        for (Map.Entry<String, Long> entry : entryList) {
            System.out.println(entry.getKey() + "," + entry.getValue());
        }
    }

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("Get CustomerProportion").getOrCreate();
        String UserInfoFileName = "./src/main/resources/data_format1/data_format1/user_info_format1.csv";
        String UserLogFileName = "./src/main/resources/data_format1/data_format1/user_log_format1.csv";

        Dataset<UserLog> UserLogDataset = spark.read().textFile(UserLogFileName)
                .map((MapFunction<String, UserLog>) UserLog::new, Encoders.bean(UserLog.class))
                .filter((FilterFunction<UserLog>) UserLog::isPurchase);

        Dataset<String> CustomerIdDataset = UserLogDataset
                .map((MapFunction<UserLog, String>) UserLog::getUserId, Encoders.STRING());

        Set<String> CustomerIdSet = new HashSet<>(CustomerIdDataset.collectAsList());

        Dataset<UserInfo> UserInfoDataset = spark.read().textFile(UserInfoFileName)
                .map((MapFunction<String, UserInfo>) UserInfo::new, Encoders.bean(UserInfo.class))
                .filter((FilterFunction<UserInfo>) userInfo -> CustomerIdSet.contains(userInfo.getUserId()));

        Map<String, Long> CustomerAgePair = UserInfoDataset.filter((FilterFunction<UserInfo>) UserInfo::legalAge)
                .javaRDD()
                .mapToPair((PairFunction<UserInfo, String, Integer>) userInfo -> new Tuple2<>(userInfo.getAgeRange(), 1))
                .countByKey();
        GetResult(CustomerAgePair);

        Map<String, Long> CustomerGenderPair = UserInfoDataset.filter((FilterFunction<UserInfo>) UserInfo::legalGender)
                .javaRDD()
                .mapToPair((PairFunction<UserInfo, String, Integer>) userInfo -> new Tuple2<>(userInfo.getGender(), 1))
                .countByKey();
        GetResult(CustomerGenderPair);


    }
}
